package com.qingcloud.df.executor.launcher;

import com.qingcloud.df.executor.service.CollectorJobScheduler;
import com.qingcloud.df.sdk.component.config.ComponentSystemConfig;
import com.qingcloud.df.sdk.component.context.CompInstContext;
import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class LocalTaskLauncher extends   TaskLauncher {
    Logger log =  LoggerFactory.getLogger(LocalTaskLauncher.class);

    @Resource
    private Scheduler scheduler;
    @Override
    public void start(CompInstContext context) throws Exception {
        ComponentSystemConfig componentSystemConfig = new ComponentSystemConfig();
        componentSystemConfig.setInstanceId(context.getId());
        componentSystemConfig.setTopic(context.getTopic());
        componentSystemConfig.setEnv(context.getEnv());
        createJob(context.getCron(), CollectorJobScheduler.class,componentSystemConfig);
        log.debug("组件实例id[" + context.getId() + "]定时任务创建成功@！");
    }

    @Override
    public void stop(CompInstContext context) throws Exception {
        //停止定时任务
        JobKey jobKey = new JobKey(context.getId()+"",context.getEnv());
        if(scheduler.checkExists(jobKey)) {
            scheduler.pauseJob(jobKey);
        }
    }

    public void deleteJob(ComponentSystemConfig componentSystemConfig) throws SchedulerException {
        JobKey jobKey = new JobKey(componentSystemConfig.getInstanceId()+"", componentSystemConfig.getEnv());
        if(scheduler.checkExists(jobKey)){
            scheduler.deleteJob(jobKey);
        }
    }

    public void triggerJob(ComponentSystemConfig componentSystemConfig) throws SchedulerException {
        JobKey jobKey = new JobKey(componentSystemConfig.getInstanceId()+"", componentSystemConfig.getEnv());
        if(scheduler.checkExists(jobKey)){
            scheduler.triggerJob(jobKey);
        }
    }

    /**
     * 创建调度任务
     * @param corn
     * @param jobClass
     * @param componentSystemConfig
     * @throws SchedulerException
     */
    private void createJob(String corn,
                           Class<? extends Job> jobClass,
                           ComponentSystemConfig componentSystemConfig) throws SchedulerException {

        JobKey jobKey = new JobKey(componentSystemConfig.getInstanceId()+"", componentSystemConfig.getEnv());
        TriggerKey triggerKey = new TriggerKey(componentSystemConfig.getInstanceId()+"", componentSystemConfig.getEnv());
        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity(componentSystemConfig.getInstanceId()+"", componentSystemConfig.getEnv())
                .startAt(DateBuilder.futureDate(1, DateBuilder.IntervalUnit.SECOND))
                .withSchedule(CronScheduleBuilder.cronSchedule(corn))
                .startNow().build();

        if(scheduler.checkExists(jobKey)){
            JobDetail jobDetail  = scheduler.getJobDetail(jobKey);
            jobDetail.getJobDataMap().put("componentSystemConfig",componentSystemConfig);
            scheduler.rescheduleJob(triggerKey,trigger);
        } else {
            JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(componentSystemConfig.getInstanceId()+"", componentSystemConfig.getEnv()).build();
            jobDetail.getJobDataMap().put("componentSystemConfig",componentSystemConfig);
            scheduler.scheduleJob(jobDetail, trigger);
        }
    }
}
